package com.diver.flinkdemo.sink;

import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.diver.flinkdemo.controller.DalDataController;
import com.diver.flinkdemo.entity.DalData;
import com.diver.flinkdemo.entity.xxt.LastLocation;
import com.diver.flinkdemo.entity.xxt.LocationInfo;
import com.diver.flinkdemo.entity.xxt.TerminalUnit;
import com.diver.flinkdemo.mapper.dorisMapper.LastLocationMapper;
import com.diver.flinkdemo.mapper.dorisMapper.LocationInfoMapper;
import com.diver.flinkdemo.mapper.mysqlMapper.TerminalUnitMapper;
import com.diver.flinkdemo.service.IRedisService;
import com.diver.flinkdemo.service.impl.RedisServiceImpl;
import com.diver.flinkdemo.utils.DynamicDataSourceUtil;
import com.diver.flinkdemo.utils.JedisUtil;
import jdk.nashorn.internal.ir.Terminal;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;

import java.util.Date;

/**
 * @author LuJW
 * @date 2023/3/29 11:28
 */
@Slf4j
public class LocationInfoSink extends RichSinkFunction<LocationInfo> {
    /**
     * sqlSessionFactory
     */
    private SqlSessionFactory sqlSessionFactory;

//    private IRedisService redisService = new RedisServiceImpl();

    @Override
    public void open(Configuration parameters) throws Exception {

        if (DynamicDataSourceUtil.dorisSessionFactory == null) {
            DynamicDataSourceUtil sourceUtil = new DynamicDataSourceUtil();
            sourceUtil.init();
            JedisUtil.poolInit();
            log.info("二次初始化完成...");
            log.info("jedis二次初始化完成...{}",JedisUtil.getJedis());
        }
        log.info("初始化获取sqlSessionFactory->{}", DynamicDataSourceUtil.dorisSessionFactory);
        sqlSessionFactory = DynamicDataSourceUtil.dorisSessionFactory;

    }

    @Override
    public void invoke(LocationInfo value, Context context) throws Exception {
        SqlSession sqlSession = null;
        LastLocation lastLocation = new LastLocation();
        try {
            //获取sqlSession
            log.info("sqlSessionFactory->{}",sqlSessionFactory);
            sqlSession = sqlSessionFactory.openSession();
            if (value!=null) {
                BeanUtils.copyProperties(lastLocation,value);
                if ("PARSED_LOCATION".equals(value.getType())) {
                    sqlSession.getMapper(LocationInfoMapper.class).insert(value);
                    log.info("LocationInfo save complete!");
                }else {
                    //不使用设备原始上报的经纬度
                    lastLocation.setLat(null);
                    lastLocation.setLon(null);
                }
                lastLocation.setCardNum(value.getImei());
                //如果类型为心跳，更新currentStatus字段（1：在线；2：离线）
                if ("HEARTBEAT".equals(value.getTypeIdent())) {
                    lastLocation.setCurrentStatus("1");
                    IRedisService redisService = new RedisServiceImpl();
                    redisService.set("heartbeat_"+value.getImei(),600L,"online");
                }
                //判断imei之前是否已在库中存在，存在更新修改时间，不存在更新创建时间
                if (sqlSession.getMapper(LastLocationMapper.class).selectById(value.getImei()) != null) {
                    lastLocation.setUpdateTime(new Date());
                    //前一步dalData转为locationInfo时设置了createTime,所以这里需要置null
                    lastLocation.setCreateTime(null);
                }else {
                    lastLocation.setCreateTime(new Date());
                }
                sqlSession.getMapper(LastLocationMapper.class).insert(lastLocation);
                log.info("LastLocation save complete!");
            }else {
                log.info("value == null,不进行入库操作");
            }
        } catch (Exception e) {
            log.error(this.getClass().getName() + ".invoke()执行失败", e);
            //重试一次
            try {
                if (sqlSession != null && value!=null) {
                    BeanUtils.copyProperties(lastLocation,value);
                    if ("PARSED_LOCATION".equals(value.getType())) {
                        sqlSession.getMapper(LocationInfoMapper.class).insert(value);
                        log.info("LocationInfo save complete!");
                    }else {
                        //不使用设备原始上报的经纬度
                        lastLocation.setLat(null);
                        lastLocation.setLon(null);
                    }
                    lastLocation.setCardNum(value.getImei());
                    //如果类型为心跳，更新currentStatus字段（1：在线；2：离线）
                    if ("HEARTBEAT".equals(value.getTypeIdent())) {
                        lastLocation.setCurrentStatus("1");
                        IRedisService redisService = new RedisServiceImpl();
                        redisService.set("heartbeat_"+value.getImei(),600L,"online");
                    }
                    //判断imei之前是否已在库中存在，存在更新修改时间，不存在更新创建时间
                    if (sqlSession.getMapper(LastLocationMapper.class).selectById(value.getImei()) != null) {
                        lastLocation.setUpdateTime(new Date());
                    }else {
                        lastLocation.setCreateTime(new Date());
                    }
                    sqlSession.getMapper(LastLocationMapper.class).insert(lastLocation);
                    log.info("LastLocation save complete!");
                }
            } catch (Exception ex) {
                log.error(this.getClass().getName() + ".invoke()重试执行失败", e);
            }
        } finally {
            //关闭sqlSession
            if (sqlSession != null) {
                log.info("session commit -> session close");
                DynamicDataSourceUtil.sessionDoAfter(sqlSession);

            }
        }
    }
}
